package com.alibaba.otter.node.etl.load.loader.mq.rabbit;


import com.alibaba.otter.node.etl.load.loader.mq.rabbit.exception.MqAckException;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.exception.MqUnexpectedException;
import com.alibaba.otter.node.etl.load.loader.mq.rabbit.model.MessageContext;
import lombok.AccessLevel;
import lombok.NoArgsConstructor;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;

@Slf4j
@NoArgsConstructor(access = AccessLevel.PRIVATE)
public class MqUtils {

    public static void sleep(long ms) {
        try {
            if (ms > 0) {
                Thread.sleep(ms);
            }
        } catch (InterruptedException e) {
            throw new MqUnexpectedException(e.getMessage(), e);
        }
    }

    public static void ack(MessageContext context) {
        if (context == null || context.getEnvelope() == null) {
            return;
        }
        long deliveryTag = context.getEnvelope().getDeliveryTag();
        try {
            context.getChannel().basicAck(deliveryTag, false);
        } catch (IOException e) {
            throw new MqAckException("消息ack出错：连接异常或远端关闭", context, e);
        }
    }

    public static void nack(MessageContext ctx) {
        if (ctx == null || ctx.getEnvelope() == null) {
            return;
        }
        long deliveryTag = ctx.getEnvelope().getDeliveryTag();
        try {
            ctx.getChannel().basicNack(deliveryTag, false, true);
        } catch (IOException e) {
            log.error("消息nack出错：连接异常或远端关闭: " + deliveryTag, e);
        }
    }

    public static void reject(MessageContext ctx) {
        if (ctx == null || ctx.getEnvelope() == null) {
            return;
        }
        long deliveryTag = ctx.getEnvelope().getDeliveryTag();
        try {
            ctx.getChannel().basicReject(deliveryTag, false);
        } catch (IOException e) {
            log.error("消息reject出错：连接异常或远端关闭: " + deliveryTag, e);
        }
    }

    public static boolean isBItNotZero(long input, int bitFromLow) {
        // 不允许负位移
        if (bitFromLow < 0) {
            return false;
        }
        return (input >>> bitFromLow & 1) == 1;
    }
}
